package com.alibaba.middleware.race.sync.util;

import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by yh on 2017/6/7.
 * |mysql-bin.00001717148759|1496736165000|middleware3|student|I|id:1:1|NULL|1|first_name:2:0|NULL|徐|last_name:2:0|NULL|依|sex:2:0|NULL|男|score:1:0|NULL|66|
 * -XX:InitialHeapSize=3221225472 -XX:MaxDirectMemorySize=209715200 -XX:MaxHeapSize=3221225472 -XX:MaxNewSize=1073741824 -XX:MaxTenuringThreshold=6 -XX:NewSize=1073741824 -XX:OldPLABSize=16 -XX:OldSize=2147483648 -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+UseCompressedOops -XX:+UseConcMarkSweepGC -XX:+UseParNewGC
 *
 */
public class TestServer {

    public static final byte SEP = '|';
    public static final byte SEP2 = ':';
    public static final byte SEPLINE = '\n';
    
    public static final byte PRAMERYUPDATE = 'P';//ID为新
    public static final byte PRAMERYUPDATE2 = 'Q';//
    
    public static final byte UPDATE = 'U';
    public static final byte INSERT = 'I';
    public static final byte DELETE = 'D';
    
    private static final String[] files = new String[]{"E:\\alibaba\\test\\canal.txt"};
//    private static final String[] files = new String[]{"1.txt","2.txt","3.txr","4.txt","5.txt",
//            "6.txt","7.txt","8.txt","9.txt","10.txt"};
    private static Map<Long, TreeMap<Long, byte[]>> serverData = new ConcurrentHashMap<>();
    private static Map<Long, ResultRs> resultRsMap = new HashMap<>();

    public static void main(String[] args) throws Exception {
        TestServer testServer = new TestServer();
        testServer.handleData(args);
    }

    private void handleData(String[] args) throws Exception {
        @SuppressWarnings("resource")
		RandomAccessFile randomAccessFile = new RandomAccessFile(files[0], "r");
        FileChannel fileChannel = randomAccessFile.getChannel();
        MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileChannel.size());
        ByteBuffer byteBuffer = mappedByteBuffer.duplicate();
        final String schemaTable = args[0]+"|"+args[1];

        long time1 = System.nanoTime();

        long timeStamp = 0;
        long primaryKey;
        long oldPrimaryKey;
        byte[] lineBytes = new byte[1024]; // 放置临时数据

        int lineBytesPosition = 0;
     

        while (byteBuffer.position() < byteBuffer.limit()) {
            lineBytesPosition = 0;
            /* 跳过最初的binaryId */
            byteBuffer.get();
            while (byteBuffer.get() != SEP) {}
            /* 计算timestamp */
            while ((lineBytes[lineBytesPosition++] = byteBuffer.get()) != SEP) {}
            timeStamp = Long.parseLong(new String(lineBytes, 0, lineBytesPosition - 1));
          

            lineBytesPosition = 0;
            /* 计算schema+table */
            while ((lineBytes[lineBytesPosition++] = byteBuffer.get()) != SEP) {}
            while ((lineBytes[lineBytesPosition++] = byteBuffer.get()) != SEP) {}
            if (!new String(lineBytes, 0, lineBytesPosition - 1).equals(schemaTable)) {
            	//行结束
                while (byteBuffer.get() != SEPLINE) {}
                continue;
            }
            
            /* 计算类型 */
            byte[] historyData = new byte[1024];
            int historyDatasPosition = 1;
            historyData[0] = byteBuffer.get();
            byteBuffer.get();
            /* 跳过主键类型 */
            while (byteBuffer.get() != SEP2) {}
            while (byteBuffer.get() != SEP2) {}
            while (byteBuffer.get() != SEP) {}
            /* 获取旧主键值 */
            lineBytesPosition = 0;
            while ((lineBytes[lineBytesPosition++] = byteBuffer.get()) != SEP) {
            }
            String pkVal = new String(lineBytes, 0, lineBytesPosition - 1);
            if (pkVal.equals("NULL")) {
                oldPrimaryKey = 0;
            } else {
                oldPrimaryKey = Long.parseLong(pkVal);
                if (historyData[0] == DELETE) {
                    if (!serverData.containsKey(oldPrimaryKey)) {
                        TreeMap<Long, byte[]> longTreeMap = new TreeMap<>();
                        longTreeMap.put(timeStamp, new byte[] {DELETE});
                        serverData.put(oldPrimaryKey, longTreeMap);
                    } else {
                        serverData.get(oldPrimaryKey).put(timeStamp, new byte[] {DELETE});
                    }
                    while (byteBuffer.get() != SEPLINE) {}
                    continue;
                }
            }
            
            /* 获取新主键值 */
            lineBytesPosition = 0;
            while ((lineBytes[lineBytesPosition++] = byteBuffer.get()) != SEP) {}
            primaryKey = Long.parseLong(new String(lineBytes, 0, lineBytesPosition - 1));
            /* 判断主键是否发生变更，若发生，则修改类型为主键变更A */
            if (primaryKey != oldPrimaryKey) {
                historyData[0] = PRAMERYUPDATE;
            }
            /* 获取后续值 */
            lineBytesPosition = 0;
            while ((historyData[historyDatasPosition++] = byteBuffer.get()) != SEPLINE) {}
            /* 存放数据 */

            // 假如发生了变更，将数据放到两个key的地方。
            if (historyData[0] == PRAMERYUPDATE) {
                byte[] oldDataBytes = new byte[1];
                oldDataBytes[0] = PRAMERYUPDATE2;
                if (!serverData.containsKey(oldPrimaryKey)) {
                    TreeMap<Long, byte[]> longTreeMap = new TreeMap<>();
                    longTreeMap.put(timeStamp, oldDataBytes);
                    serverData.put(oldPrimaryKey, longTreeMap);
                } else {
                    serverData.get(oldPrimaryKey).put(timeStamp, oldDataBytes);
                }

                byte[] newDataBytes = new byte[historyDatasPosition + 7];
                newDataBytes[0] = historyData[0];
                System.arraycopy(ServerFileUtils.long2Bytes(oldPrimaryKey), 0, newDataBytes, 1, 8);
                System.arraycopy(historyData, 1, newDataBytes, 9, newDataBytes.length - 9);
                if (!serverData.containsKey(primaryKey)) {
                    TreeMap<Long, byte[]> longTreeMap = new TreeMap<>();
                    longTreeMap.put(timeStamp, newDataBytes);
                    serverData.put(primaryKey, longTreeMap);
                } else {
                    serverData.get(primaryKey).put(timeStamp, newDataBytes);
                }
            } else {
                byte[] dataBytes = new byte[historyDatasPosition-1];
                System.arraycopy(historyData, 0, dataBytes, 0, dataBytes.length);
                if (!serverData.containsKey(primaryKey)) {
                    TreeMap<Long, byte[]> longTreeMap = new TreeMap<>();
                    longTreeMap.put(timeStamp, dataBytes);
                    serverData.put(primaryKey, longTreeMap);
                } else {
                    serverData.get(primaryKey).put(timeStamp, dataBytes);
                }
            }
        }

        long time2 = System.nanoTime();
        System.out.println("总时间：" + (time2 - time1)/1000000);

        for (Map.Entry<Long, TreeMap<Long, byte[]>> entry : serverData.entrySet()) {
            if (entry.getKey() > Long.decode(args[2]) && entry.getKey() < Long.decode(args[3])) {
                resetData(entry.getKey(), entry.getValue(), 0);
            }
        }

        for (Map.Entry<Long, ResultRs> entry : resultRsMap.entrySet()) {
            System.out.println("primary ： " + entry.getKey());
            for (Map.Entry<String, Long> entry1 : entry.getValue().longColums.entrySet()) {
                System.out.println(entry1.getKey() + " ： " + entry1.getValue());
            }
            for (Map.Entry<String, String> entry1 : entry.getValue().stringColums.entrySet()) {
                System.out.println(entry1.getKey() + " ： " + entry1.getValue());
            }
            System.out.println("____________________________");
        }
        long time3 = System.nanoTime();
        System.out.println("总时间：" + (time3 - time2)/1000000);
    }

    private static void resetData(long key, TreeMap<Long, byte[]> value, long primary)  {
        ResultRs resultRs;
        if (primary == 0) {
            resultRs = new ResultRs();
            resultRsMap.put(key, resultRs);
        } else {
            resultRs = resultRsMap.get(primary);
        }

        TreeMap<Long, byte[]> valueTree = value;
        while (valueTree.size() > 0) {
            byte[] valueData = valueTree.pollLastEntry().getValue();
            if (valueData[0] == DELETE || valueData[0] == PRAMERYUPDATE2) {
                break;
            } else if (valueData[0] == UPDATE || valueData[0] == INSERT) {
                resultRs.updateData(valueData, 1);
            } else if (valueData[0] == PRAMERYUPDATE) {
                resultRs.updateData(valueData, 9);
                byte[] oldPrimaryBytes = new byte[8];
                System.arraycopy(valueData, 1, oldPrimaryBytes, 0, 8);
                long oldPrimary = ServerFileUtils.bytes2Long(oldPrimaryBytes);
                resetData(oldPrimary, serverData.get(oldPrimary), key);//循环
            }
        }
    }
}
